package com.liuyi.netty.im.server.handle.cluster.impl;

import com.alibaba.fastjson.JSONObject;
import com.liuyi.netty.im.commons.codes.data.DataEvent;
import com.liuyi.netty.im.server.handle.cluster.ClusterMessageDisposeManage;
import com.liuyi.netty.im.server.handle.cluster.ClusterSend;
import com.liuyi.netty.im.server.handle.message.SendMessageProcessor;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
public abstract class AbstractClusterSend implements ClusterSend {

    private final SendMessageProcessor sendMessageProcessor;
    private final ThreadPoolExecutor threadPoolExecutor;

    protected AbstractClusterSend(SendMessageProcessor sendMessageProcessor, ThreadPoolExecutor threadPoolExecutor) {
        this.sendMessageProcessor = sendMessageProcessor;
        this.threadPoolExecutor = threadPoolExecutor;
    }

    @Override
    public void accept(DataEvent data) {
        try {
            List<Channel> channels = ClusterMessageDisposeManage.get(data.getToType()).dispose(data);
            if (channels != null && !channels.isEmpty()) {
                channels.forEach(channel -> {
                    if (!this.sendMessageProcessor.offer(new SendMessageProcessor.Message(channel, data))) {
                        //放入消息队列失败
                        log.error(JSONObject.toJSONString(data) + " accept fail");
                    }
                });
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

}
